package com.java.code.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.springframework.util.SerializationUtils;

import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * @author weichuanyuan
 * @e-mail wei.cy@bsoft.com.cn
 * @date 2021/10/19 21:32
 * @package com.java.code.kafka
 * @description
 */
public class MQProducerByKafka implements MQProducer {
    private Producer<String, String> kafkaProducer = null;
    private Properties properties;
    private int futureGetTimeout = 3;

    public MQProducerByKafka() {

    }

    @Override
    public void start() {
        if (this.properties != null && "false".equals(this.properties.getProperty("enabled"))) {
            return;
        }
        if (this.kafkaProducer == null) {
            this.properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            this.properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, BytesSerializer.class.getName());
            //定义Partitioner
            this.properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());
            this.kafkaProducer = new KafkaProducer(this.properties);
        }
    }

    @Override
    public void shutdown() {
        if (this.properties != null && "false".equals(this.properties.getProperty("enabled"))) {
            return;
        }
        if (this.kafkaProducer != null) {
            this.kafkaProducer.flush();
            this.kafkaProducer.close();
            this.kafkaProducer = null;
        }
    }

    @Override
    public boolean send(MQMessage<?> mqMessage) {
        byte[] bData = SerializationUtils.serialize(mqMessage);
        ProducerRecord<String, String> kafkaMessage = new ProducerRecord(this.getSendKey(mqMessage.getTopic(),
                mqMessage.getTag()), new Bytes(bData));
        try {
            Future<RecordMetadata> future = this.kafkaProducer.send(kafkaMessage);
            RecordMetadata var5 = (RecordMetadata) future.get((long) this.futureGetTimeout, TimeUnit.SECONDS);
            return true;
        } catch (Exception var6) {
            var6.printStackTrace();
            return false;
        }
    }


    private String getSendKey(String topic, String tag) {
        return tag != null && tag.trim().length() > 0 ? topic + "-" + tag : topic;
    }

    public int getFutureGetTimeout() {
        return this.futureGetTimeout;
    }

    public void setFutureGetTimeout(int futureGetTimeout) {
        this.futureGetTimeout = futureGetTimeout;
    }

    public Properties getProperties() {
        return this.properties;
    }

    public void setProperties(Properties properties) {
        this.properties = properties;
    }
}
